package com.liao.mr;

import com.liao.kv.key.ComDimension;
import com.liao.kv.key.ContactDimension;
import com.liao.kv.key.DateDimension;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.Text;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

/*
需求：某个人（一个手机号）按年-月-日查询所有通话记录（通话次数，通话时长）
一条数据对应两个联系人和三个时间维度，所以需要一条数据写入，多条数据写出
*/
public class CountDurationMapper extends TableMapper<ComDimension, Text>{

    private Map<String, String> contacts;

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        init();
    }

    //存放联系人电话与姓名的映射
    private void init(){
        contacts = new HashMap<String, String>();
        contacts.put("13520903125","周伟业");
        contacts.put("16910901085","林子明");
        contacts.put("13448821297","庄静丽");
        contacts.put("15889335163","陈小婷");
        contacts.put("17412022282","张杨平");
        contacts.put("17896343723","武冠轩");
        contacts.put("14266953292","周泉江");
        contacts.put("15215126494","余炜鹏");
        contacts.put("15203199379","张灿云");
        contacts.put("15759416288","许伟谊");
        contacts.put("19263131999","郭锦贤");
        contacts.put("16327349720","朱泽亮");
        contacts.put("13186500722","吴羽静");
        contacts.put("15587479443","张铉");
        contacts.put("14524183839","蔡迪全");
        contacts.put("18438995153","林炎秀");
        contacts.put("18479781152","梁梦怡");
        contacts.put("14365454746","刁慧彬");
        contacts.put("19126668865","陈增宏");
        contacts.put("19989730877","刘瑞雄");
        contacts.put("16903221309","刘俊声");
        contacts.put("19067315772","陈学斌");
        contacts.put("18745938848","王纯淇");
        contacts.put("16896692875","梁绍荣");
        contacts.put("13849032256","梁成鑫");
        contacts.put("18578481115","廖庭康");
        contacts.put("15806088370","张华扬");
        contacts.put("18311269351","熊欢");
        contacts.put("15225974702","马晓华");
        contacts.put("19554963714","梁夏天");
        contacts.put("13851365640","叶学斌");
        contacts.put("14306688270","麦亮森");
        contacts.put("16341910532","陈文创");
        contacts.put("14203798814","苏日锐");
        contacts.put("17180970975","庄海富");
        contacts.put("14518469456","黄宇康");
        contacts.put("19129261588","张铭");
        contacts.put("14974457759","杨晴晴");
        contacts.put("19275609227","姚能燕");
        contacts.put("15688177854","陈培凤");
        contacts.put("14998094597","杜国仲");
        contacts.put("14310022062","黄俊轩");
        contacts.put("13084986182","邓伟键");
        contacts.put("13702638948","叶钦帅");
        contacts.put("19149893646","何俊霖");
        contacts.put("16750253910","文嘉星");
        contacts.put("15292418104","王程锋");
        contacts.put("14601963538","邹子庆");
        contacts.put("17943674414","邓校君");
        contacts.put("15654757543","杨君胜");
        contacts.put("14852574413","陈煜");
        contacts.put("14622059807","罗锦坚");
        contacts.put("16703327538","黄锦晖");
        contacts.put("17503731439","吴文悦");
        contacts.put("15583578253","李伟豪");
        contacts.put("15075386058","李亮蓉");
        contacts.put("15850070766","邱泽松");
        contacts.put("18091697630","杨建民");
        contacts.put("16719852599","杨泳志");
    }

    @Override                                  //输出的Text就是通话时长
    protected void map(ImmutableBytesWritable key, Result value, Context context) throws IOException, InterruptedException {
        //        主叫           日期              被叫       次数 秒数
        //01_15837312345_2017-08-10 14:10:24_13738909097_1_0180
        String rowkey= Bytes.toString(value.getRow());//获得一行电话日志  todo

        String[] values=rowkey.split("_");//分割电话日志

        String flag=values[4];//是否建立连接
        if ("0".equals(flag)) return;//如果没有建立通话直接返回结束
        String date_time = values[2];//日期
        String duration = values[5];//通话时长（秒）
        String call1 = values[1];//打电话人1
        String call2 = values[3];//打电话人2

//        20170810 14:10:24
        int year = Integer.valueOf(date_time.substring(0, 4));//截取年
        int month = Integer.valueOf(date_time.substring(5, 7));//截取月
        int day = Integer.valueOf(date_time.substring(8, 10));//截取日

        ComDimension comDimension=new ComDimension();//common维度

        /*第一个联系人维度封装*/
        ContactDimension contactDimension1=new ContactDimension();//联系人维度
        contactDimension1.setName(contacts.get(call1));
        contactDimension1.setTelephone(call1);
        //第一个联系人按年的数据
        DateDimension dateDimensionYear = new DateDimension(year, -1, -1);//年维度
        comDimension.setContactDimension(contactDimension1);
        comDimension.setDateDimension(dateDimensionYear);
        context.write(comDimension,new Text(duration));//写出
        //第一个联系人按月的数据
        DateDimension dateDimensionMonth = new DateDimension(year, month, -1);//月维度
        comDimension.setDateDimension(dateDimensionMonth);
        context.write(comDimension,new Text(duration));//写出
        //第一个联系人按日的数据
        DateDimension dateDimensionDay = new DateDimension(year, month, day);//天维度
        comDimension.setDateDimension(dateDimensionDay);
        context.write(comDimension,new Text(duration));//写出

        /*第二个联系人维度封装*/
        ContactDimension contactDimension2=new ContactDimension();//联系人维度
        contactDimension2.setName(contacts.get(call2));
        contactDimension2.setTelephone(call2);
        //第二个联系人一年的数据
        comDimension.setContactDimension(contactDimension2);
        comDimension.setDateDimension(dateDimensionYear);
        context.write(comDimension,new Text(duration));//写出
        //第二个联系人一月的数据
        comDimension.setDateDimension(dateDimensionMonth);
        context.write(comDimension,new Text(duration));//写出
        //第二个联系人一日的数据
        comDimension.setDateDimension(dateDimensionDay);
        context.write(comDimension,new Text(duration));//写出

    }
}
